package com.wqnmdb.im.netty.dispose.impl;

import com.alibaba.fastjson.JSONObject;
import com.wqnmdb.im.domain.contants.ConstantEnum;
import com.wqnmdb.im.domain.mongo.ImAffirm;
import com.wqnmdb.im.domain.mongo.ImMessage;
import com.wqnmdb.im.domain.mongo.ImUser;
import com.wqnmdb.im.domain.netty.contants.NettyIdentifier;
import com.wqnmdb.im.domain.netty.protobuf.Message;
import com.wqnmdb.im.domain.netty.protobuf.NettyModel;
import com.wqnmdb.im.netty.data.NettyData;
import com.wqnmdb.im.netty.dispose.MsgDispose;
import com.wqnmdb.im.utils.DateUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelId;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 处理消息
 * A <=> server <=> B
 * 正常情况下,server收到A的消息后推送给B,接着B收到消息后通知server,接着server再通知A.
 * A发送消息给B后,如果A在指定时间内没有收到server的消息通知,那么A就要超时重发.
 * A的超时重发可能会导致B收到重复的消息,所以B接受消息时要进行去重.
 */
@Slf4j
@Service
public class MsgDisposeImpl extends NettyData implements MsgDispose {

    @Autowired
    private AsynAndCacheService asynAndCacheService;

    @Autowired
    private DataAndCheckService dataAndCheckService;


    /**
     * 996-认证
     */
    @Override
    public void auth(ChannelHandlerContext ctx, ImUser user) {
        Channel channel = ctx.channel();

        channel.attr(NettyIdentifier.APP_NAME).set(user.getAppName());
        channel.attr(NettyIdentifier.TIMEOUT_NUM).set(0);
        channel.attr(NettyIdentifier.AUTH_KEY).set(Boolean.TRUE);
        channel.attr(NettyIdentifier.USER_ID_KEY).set(user.getUserId());
        sendResp(ctx, dataAndCheckService.getRespModel(
                        ConstantEnum.NETTY_AUTH.getCode(), ConstantEnum.SUCCESS.getCode(), "", null));

        Runnable affirmSend = () -> {
            //确认消息补发
            List<ImAffirm> msgAffirm = dataAndCheckService.getMsgAffirm(user.getUserId(), user.getAppName());
            log.info("用户：{}.未收到的确认消息数量：{}.", user.getUserId(), msgAffirm.size());
            if (!msgAffirm.isEmpty()) {
                ChannelHandlerContext userChannel = dataAndCheckService.getUserChannel(user.getUserId(), user.getAppName());
                msgAffirm.forEach(aff -> {
                    if (aff.getStatus() == -1){
                        sendResp(userChannel, dataAndCheckService.getRespModel(
                                ConstantEnum.NETTY_RECALL.getCode(), ConstantEnum.SUCCESS.getCode(),
                                dataAndCheckService.getExtras(aff.getUuid(), aff.getStatus(), aff.getSessionId()), null));
                    }else {
                        sendResp(userChannel, dataAndCheckService.getRespModel(
                                ConstantEnum.NETTY_AFFIRM.getCode(), ConstantEnum.SUCCESS.getCode(),
                                dataAndCheckService.getExtras(aff.getUuid(), aff.getStatus(), aff.getSessionId()), null));
                    }
                    asynAndCacheService.delMsgAffirm(aff.getId(), aff.getAppName());
                });
            }
        };

        Runnable reissueSend = () -> {
            //未读消息补发
            List<ImMessage> notReceiveMsg = dataAndCheckService.getNotReceiveMsg(user.getUserId(), user.getAppName());
            log.info("用户：{}.未收到的消息数量：{}.", user.getUserId(), notReceiveMsg.size());
            if (!notReceiveMsg.isEmpty()) {
                notReceiveMsg.forEach(rMsg -> {
                    Message.MessageInfo.Builder builder = Message.MessageInfo.newBuilder();
                    BeanUtils.copyProperties(rMsg, builder);
                    sendResp(ctx, dataAndCheckService.getRespModel(
                                    ConstantEnum.NETTY_SEND.getCode(), ConstantEnum.SUCCESS.getCode(), null, builder));
                });
            }
        };
        affirmSend.run();
        reissueSend.run();
    }

    /**
     * 96-创建sessionID
     */
    @Override
    public void createSession(ChannelHandlerContext ctx, NettyModel.ReqModel reqModel, ImUser user) {
        //创建会话
        String appName = ctx.channel().attr(NettyIdentifier.APP_NAME).get();
        JSONObject extras = JSONObject.parseObject(reqModel.getExtras());
        Integer toUserId = extras.getInteger("toUserId");
        String sessionId = asynAndCacheService.getSessionId(user.getUserId(),
                toUserId, appName, asynAndCacheService.getSessionCollectionName(appName));
        if (sessionId != null) {
            extras.put("sessionId", sessionId);
            sendResp(ctx, dataAndCheckService.getRespModel(ConstantEnum.NETTY_CREATE_SESSION.getCode(),
                            ConstantEnum.SUCCESS.getCode(), extras.toJSONString(), null));
        }
    }

    /**
     * 1-pong
     */
    @Override
    public void pong(ChannelHandlerContext ctx) {
        sendResp(ctx, dataAndCheckService.getRespModel(
                        ConstantEnum.NETTY_HEARTBEAT.getCode(), ConstantEnum.SUCCESS.getCode(), "", null));
    }

    /**
     * 2-发消息 一对一
     */
    @Override
    public void singleSend(ChannelHandlerContext ctx, NettyModel.ReqModel reqModel, ImUser user) {
        //发送消息
        String appName = ctx.channel().attr(NettyIdentifier.APP_NAME).get();
        Message.MessageInfo msgInfo = reqModel.getMsgInfo();
        if (msgInfo != null) {
            Integer fromUserId = msgInfo.getFromUserId();
            Integer toUserId = msgInfo.getToUserId();
            String sessionId = msgInfo.getSessionId();
            //check fromAccount and toAccount
            if (asynAndCacheService.checkUserExists(toUserId, appName, asynAndCacheService.getUserCollectionName(appName))
                    && user.getUserId().equals(fromUserId)
                    && asynAndCacheService.checkSessionId(
                    sessionId, user.getUserId(), toUserId, asynAndCacheService.getSessionCollectionName(appName))) {
                log.info("APP名称：{}.发送消息，发送者：{}.接收者：{}.", appName, fromUserId, toUserId);
                //存消息
                asynAndCacheService.insertMsg(msgInfo, appName, asynAndCacheService.getMsgCollectionName(appName));
                //获取接收这通道
                ChannelHandlerContext userChannel = dataAndCheckService.getUserChannel(toUserId, appName);

                if (userChannel != null) {
                    sendResp(userChannel, dataAndCheckService.getRespModel(ConstantEnum.NETTY_SEND.getCode(),
                                    ConstantEnum.SUCCESS.getCode(), null, msgInfo.toBuilder()));
                }else {
                    //对方不在线，通知发送方服务端已接收消息
                    sendResp(ctx, dataAndCheckService.getRespModel(ConstantEnum.NETTY_AFFIRM.getCode(),
                            ConstantEnum.SUCCESS.getCode(),
                            dataAndCheckService.getExtras(msgInfo.getUuid(), 0, sessionId), null));
                }
            }
        }
    }

    /**
     * 3-消息确认
     */
    @Override
    public void affirm(ChannelHandlerContext ctx, NettyModel.ReqModel reqModel, ImUser user) {
        //消息确认
        String appName = ctx.channel().attr(NettyIdentifier.APP_NAME).get();
        if (StringUtils.isNotBlank(reqModel.getExtras())) {
            JSONObject extras = JSONObject.parseObject(reqModel.getExtras());
            String msgUuId = extras.getString("uuid");
            // status  -1已撤回 1-已送达  2-已读
            int status = extras.getIntValue("status");
            String sessionId = extras.getString("sessionId");

            ImMessage message = dataAndCheckService.checkMsg(
                    ctx.channel().attr(NettyIdentifier.USER_ID_KEY).get(), msgUuId, appName, sessionId, 1);
            if (message != null && asynAndCacheService.checkSessionId(
                    sessionId, user.getUserId(), message.getFromUserId(), asynAndCacheService.getSessionCollectionName(appName))) {
                log.info("APP名称：{}.消息回执，接收者：{}.消息UUID", appName, message.getFromUserId(), msgUuId);
                //更新消息状态
                if (message.getStatus() != status) {
                    asynAndCacheService.updateMsg(msgUuId, sessionId, appName, status, asynAndCacheService.getMsgCollectionName(appName));
                }
                //获取接收这通道
                ChannelHandlerContext userChannel = dataAndCheckService.getUserChannel(message.getFromUserId(), appName);
                if (userChannel != null) {
                    sendResp(userChannel, dataAndCheckService.getRespModel(
                                    ConstantEnum.NETTY_AFFIRM.getCode(), ConstantEnum.SUCCESS.getCode(),
                                    dataAndCheckService.getExtras(msgUuId, status, sessionId), null));
                }else {
                    //接收方不在线，把确认消息存起来
                    asynAndCacheService.insertMsgAffirm(message.getFromUserId(), appName, sessionId, msgUuId, status);
                }
            }
        }
    }

    /**
     * 4-撤回消息
     */
    @Override
    public void recall(ChannelHandlerContext ctx, NettyModel.ReqModel reqModel, ImUser user) {
        //撤回消息
        String appName = ctx.channel().attr(NettyIdentifier.APP_NAME).get();
        if (StringUtils.isNotBlank(reqModel.getExtras())) {
            JSONObject extras = JSONObject.parseObject(reqModel.getExtras());
            String msgUuId = extras.getString("uuid");
            String sessionId = extras.getString("sessionId");

            ImMessage message = dataAndCheckService.checkMsg(
                    ctx.channel().attr(NettyIdentifier.USER_ID_KEY).get(),  msgUuId, appName, sessionId, 2);
            if (message != null && message.getStatus() != -1) {
                Integer userId = user.getUserId().equals(message.getFromUserId()) ? message.getToUserId() : message.getFromUserId();
                //检查session
                if (asynAndCacheService.checkSessionId(sessionId, user.getUserId(),
                        user.getUserId().equals(message.getFromUserId()) ? message.getToUserId() : message.getFromUserId(),
                        asynAndCacheService.getSessionCollectionName(appName))) {
                    log.info("APP名称：{}.撤回消息，接收者：{}.消息UUID", appName, userId, msgUuId);
                    asynAndCacheService.updateMsg(msgUuId, sessionId, appName, -1, asynAndCacheService.getMsgCollectionName(appName));
                    //获取接收这通道ID
                    ChannelHandlerContext userChannel = dataAndCheckService.getUserChannel(userId, appName);
                    if (userChannel != null) {
                        sendResp(userChannel, dataAndCheckService.getRespModel(
                                        ConstantEnum.NETTY_RECALL.getCode(), ConstantEnum.SUCCESS.getCode(),
                                        dataAndCheckService.getExtras(msgUuId, -1, sessionId), null));
                    }else {
                        //通知发送方对方不在线
                        sendResp(ctx, dataAndCheckService.getRespModel(
                                ConstantEnum.NETTY_RECALL.getCode(), ConstantEnum.SUCCESS.getCode(),
                                dataAndCheckService.getExtras(msgUuId, -1, sessionId), null));
                        //接收方不在线，把确认消息存起来
                        asynAndCacheService.insertMsgAffirm(message.getFromUserId(), appName, sessionId, msgUuId, -1);
                    }
                }
            }
        }
    }


    /**
     * 消息错误
     */
    @Override
    public void msgError(ChannelHandlerContext ctx, NettyModel.ReqModel reqModel) {
        JSONObject jo = new JSONObject();
        jo.put("error", "消息错误");
        jo.put("info", "发送消息指令请求数据异常");
        sendResp(ctx, dataAndCheckService.getRespModel(
                        ConstantEnum.NETTY_ERROR.getCode(), ConstantEnum.FAIL.getCode(), jo.toJSONString(), null));

    }

    /**
     * 发送响应
     */
    @Override
    public void sendResp(ChannelHandlerContext ctx, NettyModel.RespModel builder) {
        if (ctx == null || builder == null) {
            log.info("通道不存在或消息为空");
            return;
        }
        log.info("回复通道:{}.消息:{}.", ctx.channel().id(), builder);
        //将客户端的信息直接返回写入ctx刷新缓存区
        ctx.writeAndFlush(builder);
    }
}
